package com.whoami.idmagic.hbase;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

import static com.whoami.idmagic.hbase.HFileWriterJob.CF_DEFAULT;

/**
 * @author tzp
 * @since 2019/8/15
 */
public class HFileWriterMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    private ImmutableBytesWritable rowkey = new ImmutableBytesWritable();
    private byte[] cf = CF_DEFAULT.getBytes();

    private final String[] colNames = new String[]{
            "user_no",
            "mobile_no",
            "deviceid",
            "id_type",
            "is_ps",
            "is_p18",
            "is_p24",
            "is_p36",
            "is_p18_peipao",
            "is_p24_peipao",
            "is_p36_peipao",
            "is_big",
            "sx_amt",
            "max_duotou",
            "product_code",
            "dt",
            "media",
            "jf_risk_price",
            "jf_risk_price_peipao",
            "sc_risk_price"
    };

    @Override
    protected void map(LongWritable ignore, Text value, Context context) throws IOException, InterruptedException {
        String[] cols = value.toString().split("\u0001");
        rowkey.set(cols[0].getBytes());
        Put put = new Put(cols[0].getBytes());
        for (int i = 1; i < colNames.length; i++) {
            if (StringUtils.isNotEmpty(cols[i])) {
                put.addColumn(cf, colNames[i].getBytes(), cols[i].getBytes());
            }
        }
        context.write(rowkey, put);
    }
}
